package sparksql.say8.lesson01

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SaveMode}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/5/4.
  */
object AralHotProductSpark {
  def main(args: Array[String]): Unit = {
    /**
      * 第一步 创建程序入口
      */
    val conf = new SparkConf().setMaster("local").setAppName(s"${this.getClass.getSimpleName}")
    val sc = new SparkContext(conf)
    val sqlContext = new HiveContext(sc)
    //自定义的聚合函数
    sqlContext.udf.register("group_dinstict_func",GroupDinstictUDAF)
    sqlContext.udf.register("get_product_status",(str:String) =>{//json
      var status=0;
      for( s <- str.split(",")){
        if(s.contains("product_status")){
          status=s.split(":")(1).toInt
        }
      }
    })

    /**
      * 第二步 接收外面传进来的参数
      *   起始时间 startTime
      *   结束时间 endTime
      */
    val Array(startTime,endTime)=args
    /**
      * 第三步：
      *      根据时间去Hive表里面获取用户行为的数据
      */
   val cityid2ActionRow: RDD[(Long, Row)] = getActionDataByRange(startTime,endTime,sqlContext)


    /**
      * 【接下来我做的这些事，不一定是最好的选择】，大家可以按照自己思路做。
      * 第四步：
      *       获取城市信息表（mysql）
      */
    val cityid2Cityrow: RDD[(Long, Row)] = getCityInfo(sqlContext)

    /**
      * 第五步：
      * 第三步和第四步的表进行 join （cityid）  -> table
      */
    generateTempTable(sqlContext,cityid2ActionRow,cityid2Cityrow)
    /**
      * 第六步：
      * 统计商品点击的次数
      */
    productClickCountByArea(sqlContext)
    /**
      * 第七步：
      * 上一步的结果 跟 商品信息表进行join  性能不好
      */
    genarateFullTable(sqlContext)

    /**
      * 第八步：
      * 分组求TopN
      */
    topN(sqlContext)
    /**
      * 第九步：
      * 把最后的结果写入mysql数据库
      */
    persistResult(sqlContext)
  }

  /**
    *获取用户行为数据
    * NULL ' null' ' NULL'  ' '
    * click_product_id    ' NULL'
    * * @param startTime 开始时间
    * @param endTime  结束时间
    * @param sqlContext  程序入口
    */
  def  getActionDataByRange( startTime:String,endTime:String,sqlContext:HiveContext):RDD[(Long,Row)]={
    val sql=
      s"""
         SELECT
               city_id,click_product_id
         FROM
               user_vistit_action
         WHERE
               click_product_id is not null and
               click_product_id != 'null' and
               click_product_id != 'NULL' and
                click_product_id != '' and
               date >= $startTime and
               date <= $endTime
      """
    sqlContext.sql(sql).rdd.map( row => (row.getLong(0),row))
  }

  /**
    * 获取城市信息表
    * @param sqlContext 程序入口
    * @return k:城市id  v：城市信息
    *         mysql
    *  URL:
    *  tableName
    *  username
    *  password
    */
  def getCityInfo(sqlContext:HiveContext):RDD[(Long,Row)]={
    val df: DataFrame = sqlContext.read.format("jdbc")
      .option("url", "jdbc:mysql://hadoop1:3306/aura")
      .option("user", "aura")
      .option("password", "aura")
      .option("dbtable", "city_info")
      .load()
     df.rdd.map( row => (row.getLong(0),row))
  }

  /**
    * 用户行为表 join 城市信息表
    * @param sqlContext 程序入口
    * @param cityid2ActionRow  用户行为表
    * @param cityid2Cityrow  城市信息表
    */
  def generateTempTable(sqlContext:HiveContext, cityid2ActionRow: RDD[(Long, Row)],
                        cityid2Cityrow: RDD[(Long, Row)]):Unit={
    /**
      * Long cityid
      * row  actionRow
      * row  cityinforRow
      */
    val resultRDD: RDD[(Long, (Row, Row))] = cityid2ActionRow.join(cityid2Cityrow)

    val rowRDD: RDD[Row] = resultRDD.map(tuple => {
      //cityid
      val city_id = tuple._1.toLong
      //cityid,click_product_id
      val actionRow = tuple._2._1
      //cityid,city_name,area
      val cityinfoRow = tuple._2._2

      val click_product_id = actionRow.getLong(1)
      val city_name = cityinfoRow.getString(1)
      val area = cityinfoRow.getString(2)
      Row(city_id, click_product_id, city_name, area)
    })

    val schema=StructType(
      StructField("city_id",LongType,true)::
      StructField("click_product_id",LongType,true)::
      StructField("city_name",StringType,true)::
      StructField("area",StringType,true)::Nil
    )

    val df: DataFrame = sqlContext.createDataFrame(rowRDD,schema)
    df.createOrReplaceTempView("action_city")
  }

  /**
    * 统计每个商品在各区域里面点击的次数
    *
    * 这些数据的区域里面包含哪些城市
    *
    *华北地区：
    * 天津
    * 北京
    *
    * @param sqlContext
    */
  def productClickCountByArea(sqlContext:HiveContext): Unit ={

    /**
      *
      *定义一个聚合函数：
      * select city_id,city_name from action_city
      * 1:beijing
      * 2:tianjing    =>  1:bejing,2:tianjing,3:shanghai
      * 1:beiing
      * 3:shanghai
      *
      * val str=""
      *
      * 1:beijing,2:tianjing,3:shanghai
      *
      */

    val sql=
      """
         select
              area,click_product_id,count(*) click_count，
              group_dinstict_func(concat_ws(':',city_id,city_name)) city_infos
               from
         action_city
               group by
         area,click_product_id
      """

    sqlContext.sql(sql).createOrReplaceTempView("product_aread_count")

  }

  /**
    * 上一次的结果跟商品信息表进行join
    * @param sqlContext
    */
  def genarateFullTable(sqlContext:HiveContext): Unit ={
      val sql=
        """
            select
                 pac.area,
                 pac.click_product_id,
                 pac.click_count,
                 pac.city_infos,
                 pi.product_name,
                 if(get_product_status(pi.extends_info)= 0 ,'自营商品','第三方商品') product_status
            from
                product_aread_count pac
            join
                product_info  pi
            on
                pac.click_product_id = pi.product_id

        """

      sqlContext.sql(sql).createOrReplaceTempView("area_product_full_table")

  }

  def topN(sqlContext:HiveContext): Unit ={
    val sql=
      """
         SELECT
               CASE
                   WHEN area='华北' or area='华东' then 'A级'
                   WHEN area='华南' or area='华中' then 'B级'
                   WHEN area='西北' or area='西南' then 'C级'
                   WHEN area='东北' then 'D级'
                   ELSE 'E级'
               END area_level,
               area,
               click_product_id,click_count,city_infos,product_name,product_status
         FROM
         (
         SELECT
               area, click_product_id,click_count,city_infos,product_name,product_status,
               ROW_NUMBER() OVER(PARTITION BY area ORDER BY click_count desc ) rank
         FROM
              area_product_full_table
         ) tmp
         WHERE
              tmp.rank <= 10
      """
    sqlContext.sql(sql).createOrReplaceTempView("result")
  }

  def persistResult(sqlContext:HiveContext): Unit ={
    val df: DataFrame = sqlContext.read.table("result")
    /**
      * 方式一：
      *    DataFrame  -> jdbc
      */
    df.write.mode(SaveMode.Overwrite).format("jdbc")
      .option("","")
      .option("","")
      .option("","")
      .option("","")
    /**
      * 方式二：
      *   DataFrame  -> rdd
      */
    df.rdd.foreach( row =>{
      val city_levbel = row.getString(0)
      //这种方式非要做其实也是可以的。
    })



  }

}
